package com.windy.kafka.trace.interceptor;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.windy.common.transmit.TransmitHeader;
import com.windy.kafka.trace.consumer.MessageConsumerProcessor;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Arrays;
import java.util.List;

/**
 * Created by liyufeng on 2022/6/3
 */
public class MessageListenerMethodInterceptor implements MethodInterceptor {

    private static final String LISTENER_METHOD_NAME = "onMessage";

    private TransmitHeader transmitHeader;

    private List<MessageConsumerProcessor> messageConsumerProcessors;

    public MessageListenerMethodInterceptor(TransmitHeader transmitHeader, List<MessageConsumerProcessor> messageConsumerProcessors) {
        this.transmitHeader = transmitHeader;
        this.messageConsumerProcessors = messageConsumerProcessors;
    }

    @Nullable
    @Override
    public Object invoke(@Nonnull MethodInvocation invocation) throws Throwable {

        if (!StrUtil.equals(LISTENER_METHOD_NAME, invocation.getMethod().getName())) {
            return invocation.proceed();
        }
        Object[] arguments = invocation.getArguments();
        ConsumerRecord consumerRecord = Arrays.stream(arguments)
                .filter(e -> e instanceof ConsumerRecord)
                .map(e -> (ConsumerRecord) e)
                .findFirst()
                .get();
        if (ObjectUtil.isNull(consumerRecord)) {
            return invocation.proceed();
        }
        try {
            transmitHeader.transmitHeader(() -> consumerRecord.headers());
            messageConsumerProcessors.forEach(messageConsumerProcessor -> {
                messageConsumerProcessor.beforeProcessMessage(consumerRecord);
            });
            return invocation.proceed();

        } catch (Exception e) {
            //todo 统一异常处理
            System.out.println("--- 发生消费异常 ---");
            System.out.println(e);
            //不能继续往上抛了
            //throw  new Exception(e);
        } finally {
            messageConsumerProcessors.forEach(messageConsumerProcessor -> {
                messageConsumerProcessor.afterProcessorMessage(consumerRecord);
            });

        }
    }
}
